package com.shiku.delayjob.consumer;

import com.alibaba.fastjson.JSON;
import com.shiku.delayjob.constants.DelayJobConstant;
import com.shiku.delayjob.handler.DelayJobHanler;
import com.shiku.delayjob.model.DelayJob;
import com.shiku.delayjob.model.DelayJobStatus;
import com.shiku.delayjob.model.ScoredSortedItem;
import com.shiku.delayjob.repository.DelayJobDO;
import com.shiku.delayjob.repository.DelayJobRepository;
import com.shiku.mongodb.utils.BeanCopyUtils;
import com.shiku.utils.DateUtil;
import com.shiku.utils.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RScoredSortedSet;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.TimeUnit;

@Slf4j
public abstract class AbstractDelayBucketHandler implements DelayJobHanler, Runnable {

    protected abstract RedissonClient getRedissonClient();

    protected RScoredSortedSet<ScoredSortedItem> scoredSorteSet;

    @Autowired
    protected DelayJobRepository delayJobRepository;

    @Autowired
    protected IExecuteDealyJob executeDealyJob;

    private  String getDelayBucketKey() {
        return DelayJobConstant.DELAY_QUEUE_KEY_PREFIX;
    }

    @Override
    public void run() {
        try {
             scoredSorteSet = getRedissonClient().getScoredSortedSet(getDelayBucketKey());
            log.info("定期拉取延时任务开启,刚启动先休息10秒");
            TimeUnit.SECONDS.sleep(10L);
        } catch (InterruptedException e) {
            log.error(e.getMessage());
        }

        while (true){

            try {
                pullDelayJob();
            } catch (Exception e) {
                log.error(e.getMessage(),e);
            }
        }

    }


    private void sleep(long sleep){
        try {
            TimeUnit.SECONDS.sleep(sleep);
        }catch (InterruptedException e){
            log.error("",e);
        }
    }

    @Override
    public void pullDelayJob() {

        ScoredSortedItem item = scoredSorteSet.first();
        log.debug("pullDelayJob ");
        /**
         * 没有任务
         */
        if(null==item){
            sleep(5);
            log.debug("pullDelayJob sleep 5");
            return;
        }
        /**
         * 延时时间没有到
         */
        if (item.getDelayTime() > DateUtil.currentTimeSeconds()) {
            log.debug("pullDelayJob sleep 2");
            sleep(2);
            return;
        }
        if(StringUtil.isEmpty(item.getJobId())||0==item.getDelayTime()){
            log.error("数据异常 {}", JSON.toJSONString(item));
            return;
        }
        DelayJobDO delayJob = delayJobRepository.getDelayJobByJobId(item.getJobId());
        if(null==delayJob){
            log.info("延时任务数据库不存在 移除任务 ==> jobId {} delayTime {} ",
                    item.getJobId(),item.getDelayTime());
            scoredSorteSet.remove(item);
            return;
        }else if (delayJob.getStatus()== DelayJobStatus.CANCEL.getType()||
                delayJob.getStatus()>DelayJobStatus.WAITEXEC.getType()) {
            log.info("任务状态异常,取消执行 status {} jobId {}",
                    DelayJobStatus.getAddTypeDesc(delayJob.getStatus()),item.getJobId());

            //删除旧的
            scoredSorteSet.remove(item);
            //重新计算延迟时间
            //scoredSorteSet.add(delayJob.getExecTime(),new ScoredSortedItem(delayJob.getJobId(),delayJob.getExecTime()));

        }else if (delayJob.getExecTime() > DateUtil.currentTimeSeconds()) {
            log.info("任务执行时间不对,重新计算 db time {} redis time {}  jobId {}",
                    delayJob.getExecTime(),item.getDelayTime(),item.getJobId());

            //删除旧的
            scoredSorteSet.remove(item);
            //重新计算延迟时间
           scoredSorteSet.add(delayJob.getExecTime(),new ScoredSortedItem(delayJob.getJobId(),delayJob.getExecTime()));

        } else {
            log.info("任务执行时间已到 执行任务 topic {} jobId {} execTime {}",
                    delayJob.getTopic(),delayJob.getJobId(),delayJob.getExecTime());

            DelayJob execjob=new DelayJob();
            BeanCopyUtils.copyProperties(delayJob,execjob);
            executeDealyJob.execute(execjob);

            scoredSorteSet.remove(item);
            delayJobRepository.deleteDelayJobAndTopic(delayJob.getJobId(),delayJob.getTopic());
        }



    }
}
